package other.one.advance;

import java.util.Collection;
import java.util.HashSet;
import java.util.Set;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import java.util.concurrent.PriorityBlockingQueue;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * 进阶版
 */
public class RoomServiceImplAdvance2 {
    private static final ConcurrentMap<String, Set<String>> roomCache = new ConcurrentHashMap<>();

    // 定义一个优先队列来解决进入房间和退出房间先后不确定问题
    private static final PriorityBlockingQueue<RoomRequest> queue = new PriorityBlockingQueue<>();
    private static int lastSerialNo = 0;
    public Thread thread;

    /**
     * 在构造方法内开启子线程
     */
    public RoomServiceImplAdvance2() {
        thread = new Thread(() -> {
            try {
                queueHandler();
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        });
        thread.start();
    }


    public void operation(RoomRequest roomRequest) throws RuntimeException {
        // 特殊情况判断
        if (roomRequest == null || roomRequest.getRoomId() == null || roomRequest.getUserId() == null
                || roomRequest.getSerialNo() == null || roomRequest.getSerialNo() <= 0) {
            return;
        }
        // 为房间设置默认空集合，方面后面的逻辑处理
        roomCache.putIfAbsent(roomRequest.getRoomId(), new HashSet<>());
        // 将操作放入队列
        queue.offer(roomRequest);
    }

    public Collection<String> queryUser(String roomId) throws RuntimeException {
        Set<String> userId = roomCache.get(roomId);
        return null == userId ? new HashSet<>() : userId;
    }

    /**
     * 使用子线程开启异步处理器
     *
     * @throws InterruptedException
     */
    private void queueHandler() throws InterruptedException {
        // 一直循环处理队列里的操作
        while (true) {
            // 休眠100毫秒，避免一直没有收到连续的序列号造成死循环，CPU过高
            Thread.sleep(100);
            // 获取优先队列里的数据，如果没有则阻塞
            RoomRequest take = queue.take();
            // 如果序列号不连续则丢进队列重新判断
            if (take.getSerialNo() != (lastSerialNo + 1)) {
                queue.offer(take);
                // todo 是否需要添加一个容错判断，如果连续10次都是该序列，即一秒内都没有收到代表丢失了，将序列号向下走
                continue;
            }
            switch (take.getType()) {
                case 1: {
                    joinRoom(take.getUserId(), take.getRoomId());
                    break;
                }
                case 0: {
                    leaveRoom(take.getUserId(), take.getRoomId());
                    break;
                }
            }
            // 序列号更新
            lastSerialNo = take.getSerialNo();
        }
    }

    private void joinRoom(String userId, String roomId) {
        roomCache.get(roomId).add(userId);
    }

    private void leaveRoom(String userId, String roomId) throws RuntimeException {
        Set<String> userSet = roomCache.get(roomId);
        if (userSet == null) {
            return;
        }
        userSet.remove(userId);
    }



}